package com.gomro.galaxy.foundation.bi.engine.util;

import cn.gomro.cloud.core.common.utils.DateUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gomro.galaxy.core.commons.elastic.ElasticSearchRest;
import com.gomro.galaxy.foundation.bi.domain.model.AcquireGoods;
import com.gomro.galaxy.foundation.bi.exception.BiException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.ShardSearchFailure;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.NonTransientResourceException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.support.IteratorItemReader;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.*;
import java.util.function.BiFunction;

/**
 * @Author: zhoujian
 * @Date: 2020/4/26 9:26
 * @Version 1.0
 */
@Slf4j
public class EngineUtils {
    public static final int MAX_BATCH_NUM = 300;

    public static <T> ItemReader<T> createCacheReader(Map<String, T> cache) {
        Iterator<T> iterator = getIterator(cache);
        return new IteratorItemReader<T>(iterator);
    }

    private static <T> Iterator<T> getIterator(Map<String, T> cache) {
        if (CollectionUtils.isEmpty(cache)) {
            return Collections.emptyIterator();
        } else {
            return cache.values().iterator();
        }
    }

    /**
     * 限制每批处理数量小于300
     *
     * @return
     */
    public static Date getEndDate(Date start, BiFunction<Date, Date, Long> function) {
        Date end = null;
        Date now = new Date();
        Long count = function.apply(start, null);
        log.warn("[BiEngineAnalysis-EndDate] first --->  start:{} count:{}", DateUtils.dateToStr(start), count);
        while (count > MAX_BATCH_NUM) {
            if (Objects.isNull(end) && MAX_BATCH_NUM < 1000) {
                end = DateUtils.firstDayOfNextMonth(start, 1);
            } else if (Objects.isNull(end) && MAX_BATCH_NUM >= 1000) {
                end = DateUtils.firstDayOfNextMonth(start, 1);
            }
            if (end.after(now)) {
                end = now;
            }
            count = function.apply(start, end);
            if (count <= MAX_BATCH_NUM) {
                log.warn("[BiEngineAnalysis-EndDate] update ---> start:{} end:{} count:{}", DateUtils.dateToStr(start), DateUtils.dateToStr(end), count);
                return end;
            }
            if (DateUtils.addDays(start, Calendar.DAY_OF_WEEK + 1).before(end)) {
                end = DateUtils.beforeNDay(end, Calendar.DAY_OF_WEEK);
            } else if (DateUtils.addDays(start, 2).compareTo(end) >= 0) {
                log.warn("[BiEngineAnalysis-EndDate] update ---> start:{} end:{} count:{}", DateUtils.dateToStr(start), DateUtils.dateToStr(end), count);
                return end;
            } else {
                end = DateUtils.beforeNDay(end, 1);
            }
        }
        return null;
    }

    public static <T> List<T> scrollSearch(SearchResponse response, ElasticSearchRest rest, ObjectMapper mapper,
                                                 TimeValue timeValue, Class<T> aClass) {
        String scrollId = response.getScrollId();
        if (Objects.isNull(scrollId)) {
            throw new BiException("未获取到scrollId!");
        }
        LinkedList<T> list = new LinkedList<>();
        try {
            while (response.getHits() != null && response.getHits().getHits().length != 0) {
                for (SearchHit hit : response.getHits()) {
                    list.add(mapper.readValue(hit.getSourceAsString(), aClass));
                }
                response = rest.searchScroll(scrollId, timeValue);
            }
            ShardSearchFailure[] shardFailures = response.getShardFailures();
            if ((Objects.nonNull(shardFailures) && shardFailures.length != 0) || response.isTimedOut()) {
                throw new BiException("数据查询失败 " + Arrays.toString(shardFailures));
            }
        } catch (BiException | IOException e) {
            throw new BiException(ExceptionUtils.getMessage(e));
        } finally {
            boolean scrollClear = rest.searchScrollClear(scrollId);
            if (!scrollClear) {
                log.error("[BiEngineAnalysis-scrollSearch] scrollId清除失败 {}", scrollId);
            }
        }
        return list;
    }
}
